"""
股票财务分析系统

本系统提供以下核心功能：
1. 从多个数据源获取股票财务数据
2. 对财务数据进行清洗、验证和格式化
3. 使用大模型进行专业财务分析
4. 生成结构化的投资分析报告

主要类及功能：
1. StockFinancialAnalyzer: 主分析器类
   - get_financial_data(): 获取并格式化财务数据
   - analyze_financial_data(): 使用AI模型分析财务数据
   - resolve_stock_code(): 解析股票代码/名称
   - get_current_stock_price(): 获取实时股价

2. AkshareFinancialDataFetcher: 财务数据获取类
   - get_financial_data(): 从多个数据源获取并整合财务数据
   - _get_sina_data(): 从新浪财经获取数据
   - _get_ths_data(): 从同花顺获取数据
   - _get_em_data(): 从东方财富获取数据
   - _get_ths_key_indicators(): 获取关键财务指标

输入输出说明：
StockFinancialAnalyzer.get_financial_data(stock_input: str) -> str
  输入: 股票代码(6位)或股票名称
  输出: 格式化后的财务数据文本

StockFinancialAnalyzer.analyze_financial_data(prompt_template: str) -> str
  输入: 可选的自定义提示模板
  输出: AI生成的财务分析报告(字符串格式)

AkshareFinancialDataFetcher.get_financial_data() -> dict
  输出: 结构化财务数据字典，包含:
    - 公司基本信息
    - 财务报告(按报告期)
    - 主要股东信息
"""

import os
import datetime
import pandas as pd
import akshare as ak
import json
import logging
import numpy as np
from openai import OpenAI
from datetime import datetime, timedelta

# 设置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)

# 默认提示模板路径
DEFAULT_PROMPT_TEMPLATE_PATH = "wxcloudrun/prompt/inancial_analysis_prompt_template.txt"

class StockFinancialAnalyzer:
    """
    股票财务分析器主类
    
    功能：
    - 获取并格式化财务数据
    - 使用AI模型进行财务分析
    - 股票代码解析和股价获取
    
    属性:
    financial_data: 格式化后的财务数据文本
    analysis_result: AI分析结果
    stock_price: 当前股票价格
    """
    
    def __init__(self, api_key=None, prompt_template_path=None):
        """
        初始化分析器
        
        参数:
        api_key: OpenAI API密钥
        prompt_template_path: 提示模板文件路径
        """
        logger.info("初始化StockFinancialAnalyzer...")
        if api_key is None:
            logger.info("尝试从环境变量获取API密钥")
            api_key = os.getenv("DASHSCOPE_API_KEY")
            if not api_key:
                logger.warning("未在环境变量中找到DASHSCOPE_API_KEY")

        logger.info("创建OpenAI客户端")
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
        )

        self.financial_data = None
        self.analysis_result = None
        self.stock_price = None
        self.prompt_template_path = prompt_template_path or DEFAULT_PROMPT_TEMPLATE_PATH
        logger.info("StockFinancialAnalyzer初始化完成")

    def get_financial_data(self, stock_input):
        """
        获取并格式化财务数据
        
        参数:
        stock_input: 股票代码(6位)或股票名称
        
        返回:
        格式化后的财务数据文本
        """
        logger.info(f"开始获取财务数据，输入: {stock_input}")
        try:
            logger.info(f"解析股票代码: {stock_input}")
            stock_code = self.resolve_stock_code(stock_input)
            if not stock_code:
                logger.error(f"未找到匹配的股票: {stock_input}")
                return "未找到匹配的股票"
            logger.info(f"解析成功: {stock_input} -> {stock_code}")

            logger.info(f"获取当前股价: {stock_code}")
            self.get_current_stock_price(stock_code)
            if self.stock_price:
                logger.info(f"当前股价: {self.stock_price}")
            else:
                logger.warning("未能获取当前股价")

            logger.info("创建AkshareFinancialDataFetcher实例")
            fetcher = AkshareFinancialDataFetcher(stock_code, self.stock_price)

            logger.info("获取财务数据...")
            financial_dict = fetcher.get_financial_data()
            logger.info(f"获取到财务数据，包含 {len(financial_dict.get('财务报告', {}))} 个报告期")

            logger.info("格式化财务数据...")
            formatted_data = self.format_financial_data(financial_dict)
            logger.info(f"格式化完成，文本长度: {len(formatted_data)} 字符")

            self.financial_data = formatted_data
            logger.info("财务数据准备完成")
            return self.financial_data

        except Exception as e:
            logger.exception(f"获取财务数据时出错: {str(e)}")
            return f"获取数据时出错: {str(e)}"

    def analyze_financial_data(self, stock_input, prompt_template=None):
        """
        使用AI模型分析指定股票的财务数据
        
        参数:
        stock_input: 股票代码(6位)或股票名称
        prompt_template: 可选的自定义提示模板
        
        返回:
        AI生成的财务分析报告(字符串格式)
        """
        logger.info(f"开始分析股票财务数据: {stock_input}")
        try:
            # 获取财务数据
            logger.info(f"获取财务数据: {stock_input}")
            financial_data_text = self.get_financial_data(stock_input)
            
            if "出错" in financial_data_text or "未找到" in financial_data_text:
                logger.error(f"无法获取财务数据: {financial_data_text}")
                return financial_data_text
            
            # 读取提示模板
            if prompt_template is None:
                prompt_template = self._get_default_prompt_template()
                """
	        logger.info(f"从文件读取提示模板: {self.prompt_template_path}")
                try:
                    with open(self.prompt_template_path, "r", encoding="utf-8") as f:
                        prompt_template = f.read()
                    logger.info("提示模板读取成功")
                except Exception as e:
                    logger.error(f"读取提示模板失败: {str(e)}，使用默认模板")
                    prompt_template = self._get_default_prompt_template()
	        """
            else:
                logger.info("使用传入的提示模板")
            
            full_prompt = prompt_template.format(financial_data=financial_data_text)
            logger.debug(f"完整提示词长度: {len(full_prompt)} 字符")
            
            messages = [
                {'role': 'system', 'content': '你是一位专业的金融分析师'},
                {'role': 'user', 'content': full_prompt}
            ]
            
            logger.info("调用OpenAI API进行分析...")
            try:
                completion = self.client.chat.completions.create(
                    model="qwen3-235b-a22b",
                    messages=messages,
                    extra_body={"enable_thinking": False},
                )
                logger.info("API调用成功")
                return completion.choices[0].message.content
                
            except Exception as e:
                logger.exception(f"调用OpenAI API失败: {str(e)}")
                return f"分析失败: {str(e)}"
            
        except Exception as e:
            logger.exception(f"分析财务数据时出错: {str(e)}")
            return f"分析过程中出错: {str(e)}"

    def resolve_stock_code(self, stock_input):
        """
        解析股票代码或名称
        
        参数:
        stock_input: 股票代码(6位)或股票名称
        
        返回:
        标准股票代码(带市场前缀)
        """
        logger.info(f"解析股票标识: {stock_input}")
        if isinstance(stock_input, str) and len(stock_input) == 6 and stock_input.isdigit():
            logger.debug(f"处理6位数字股票代码: {stock_input}")
            if stock_input.startswith(('6', '9')):
                resolved_code = 'sh' + stock_input
            else:
                resolved_code = 'sz' + stock_input
            logger.info(f"解析为: {resolved_code}")
            return resolved_code
        elif isinstance(stock_input, str) and len(stock_input) == 8 and stock_input.startswith(('sh', 'sz')):
            logger.info(f"输入已经是标准代码: {stock_input}")
            return stock_input
        else:
            logger.info(f"按股票名称查询: {stock_input}")
            try:
                stock_df = ak.stock_info_a_code_name()
                match = stock_df[stock_df['name'] == stock_input]
                if not match.empty:
                    code = match.iloc[0]['code']
                    if code.startswith(('6', '9')):
                        resolved_code = 'sh' + code
                    else:
                        resolved_code = 'sz' + code
                    logger.info(f"名称匹配成功: {stock_input} -> {resolved_code}")
                    return resolved_code
                logger.warning(f"未找到匹配的股票名称: {stock_input}")
                return None
            except Exception as e:
                logger.error(f"解析股票代码出错: {str(e)}")
                return None

    def get_current_stock_price(self, stock_code):
        """
        获取当前股票价格
        
        参数:
        stock_code: 标准股票代码(带市场前缀)
        """
        logger.info(f"获取当前股价: {stock_code}")
        try:
            code = stock_code[2:]
            df = ak.stock_zh_a_spot_em()
            stock_data = df[df['代码'] == code]
            if not stock_data.empty:
                self.stock_price = float(stock_data.iloc[0]['最新价'])
                logger.info(f"获取当前股价成功: {code} - {self.stock_price}")
            else:
                logger.warning(f"未找到股票价格: {code}")
                self.stock_price = None
        except Exception as e:
            logger.error(f"获取股票价格失败: {str(e)}")
            self.stock_price = None

    def format_financial_data(self, financial_dict):
        """
        格式化财务数据为文本
        
        参数:
        financial_dict: 财务数据字典
        
        返回:
        格式化后的文本
        """
        output = []

        output.append("公司基本信息")
        output.append("============")
        output.append(f"股票代码: {financial_dict.get('股票代码', '未知')}")
        basic_info = financial_dict.get("公司基本信息", {})
        output.append(f"主营业务: {basic_info.get('主营业务', '未知')}")
        output.append(f"主要产品: {basic_info.get('主要产品', '未知')}")
        output.append("")

        output.append("财务数据报告")
        output.append("============")
        report_data = financial_dict.get("财务报告", {})
        if not report_data:
            output.append("  无可用财务数据")
        else:
            sorted_dates = sorted(report_data.keys(), reverse=True)
            logger.info(f"格式化 {len(sorted_dates)} 个报告期数据")

            for date in sorted_dates:
                indicators = report_data[date]
                report_type = self._get_report_type(date)
                output.append(f"报告日期: {date} ({report_type})")

                for key in ["营业总收入", "净利润", "归属于母公司净利润", "每股收益",
                            "总资产", "总负债", "净资产", "资产负债率"]:
                    if key in indicators:
                        value = indicators[key]
                        value_str = self.format_value(value, key)
                        output.append(f"    {key}: {value_str}")

                for key in ["货币资金", "应收账款", "应付账款", "存货", "合同负债",
                            "流动比率", "速动比率", "应收账款周转天数", "存货周转天数", "总资产周转率"]:
                    if key in indicators:
                        value = indicators[key]
                        value_str = self.format_value(value, key)
                        output.append(f"    {key}: {value_str}")

                for key in ["经营活动现金流量净额", "投资活动现金流量净额", "筹资活动现金流量净额",
                            "现金净增加额", "自由现金流"]:
                    if key in indicators:
                        value = indicators[key]
                        value_str = self.format_value(value, key)
                        output.append(f"    {key}: {value_str}")

                for key in ["净资产收益率", "销售毛利率", "销售净利率"]:
                    if key in indicators:
                        value = indicators[key]
                        value_str = self.format_value(value, key)
                        output.append(f"    {key}: {value_str}")

                for key in ["股息支付率", "股息收益率"]:
                    if key in indicators:
                        value = indicators[key]
                        value_str = self.format_value(value, key)
                        output.append(f"    {key}: {value_str}")

                output.append("")

        output.append("主要股东")
        output.append("========")
        holders_data = financial_dict.get("主要股东", {})
        if not holders_data:
            output.append("  无主要股东数据")
        else:
            sorted_dates = sorted(holders_data.keys(), reverse=True)
            logger.info(f"格式化 {len(sorted_dates)} 个股东报告期")

            for date in sorted_dates:
                report_type = self._get_report_type(date)
                output.append(f"报告日期: {date} ({report_type})")
                for holder in holders_data[date]:
                    name = holder.get("股东名称", "未知")
                    share = holder.get("持股比例(%)", "未知")
                    output.append(f"  {name}: {share}%")
                output.append("")

        formatted_text = "\n".join(output)
        logger.info(f"格式化完成，文本长度: {len(formatted_text)} 字符")
        return formatted_text

    def _get_report_type(self, date_str):
        """根据日期字符串判断报告类型"""
        try:
            month_day = date_str[4:]
            if month_day == "0331":
                return "一季报"
            elif month_day == "0630":
                return "半年报"
            elif month_day == "0930":
                return "三季报"
            elif month_day == "1231":
                return "年报"
            else:
                if len(date_str) == 8:
                    date_obj = datetime.strptime(date_str, "%Y%m%d")
                    month = date_obj.month
                    day = date_obj.day
                    if month == 3 and day == 31:
                        return "一季报"
                    elif month == 6 and day == 30:
                        return "半年报"
                    elif month == 9 and day == 30:
                        return "三季报"
                    elif month == 12 and day == 31:
                        return "年报"
        except:
            pass
        return "财务报告"

    def format_value(self, value, key):
        """格式化财务数值"""
        if value is None:
            return "无数据"

        if key in ["资产负债率", "净资产收益率", "销售毛利率", "销售净利率",
                   "股息支付率", "股息收益率", "总资产周转率"]:
            return f"{value*100:.2f}%"

        if key in ["应收账款周转天数", "存货周转天数"]:
            return f"{value:.1f}天"

        if key in ["流动比率", "速动比率"]:
            return f"{value:.2f}"

        if isinstance(value, (int, float)):
            if abs(value) >= 1e8:
                return f"{value/1e8:.2f}亿"
            elif abs(value) >= 1e4:
                return f"{value/1e4:.2f}万"
            return f"{value:.2f}"

        return str(value)
    
    def _get_default_prompt_template(self):
        """获取默认提示模板"""
        return """
你是一位顶级财务分析师，需用通俗语言解读财报。请根据以下财务数据生成报告：
{financial_data}

### 报告结构：
**🔍 一、一句话筱总结**  
概况公司价值：  
"**[核心标签]型公司**"（示例：现金奶牛/成长新星/转型阵痛）  
► 核心竞争力：[20字内优势总结]  
► 主要风险点：[20字内风险提示]  

**📊 二、关键指标透视**  
1. **赚钱能力** → 毛利率[变动方向]，[白话解读]  
2. **现金健康** → 经营现金流[支撑/拖累]利润  
3. **债务安全** → 速动比率[数值][优/劣]于同行  

**🚨 三、风险雷达**  
• 当检测到异常数据时 → **"[具体风险描述]"**  
（示例：应收账款增速显著高于营收 → **"回款速度恶化！可能牺牲现金流换销售"**）  

**👑 四、股东密码**  
• 大股东动作 → [增减持比例]变化（[行为推测]）  
• 股权集中度 → [前三大股东占比]% ([影响说明])  

**🌍 五、行业定位仪**  
• 增速对比 → 公司近期增速[高于/低于]行业水平  
• 未来导航 → 面临[最大机遇]/[最大挑战]  

**💡 六、经营推理**  
"数据暗示[关键现象] → **可能**源于[合理推测原因]"  
（示例："存货量骤增 → **可能**为新产品上市备货"）  

---
✅ **输出规范**：  
1. 用**加粗**标记核心结论，🔴标红≤3处重大风险  
2. 每个模块用图标引导（🔍📊🚨👑🌍💡）  
3. 适当的留空行和分割线, 让阅读更轻松一些
"""


class AkshareFinancialDataFetcher:
    """
    财务数据获取类
    
    功能:
    - 从多个数据源获取财务数据
    - 数据清洗和验证
    - 计算衍生财务指标
    
    属性:
    stock_code: 股票代码
    stock_price: 当前股价
    """
    
    def __init__(self, stock_code, stock_price=None):
        logger.info(f"初始化AkshareFinancialDataFetcher: {stock_code}")
        self.stock_code = stock_code
        self.market_prefix = stock_code[:2].lower()
        self.stock_number = stock_code[2:]
        self.ths_symbol = self.stock_number
        self.em_symbol = f"{self.market_prefix}{self.stock_number}"
        self.stock_price = stock_price

        self.core_indicators = {
            "营业总收入": ["sina_营业总收入", "ths_营业总收入", "em_TOTALOPERATEREVE"],
            "净利润": ["sina_净利润", "ths_净利润", "em_NETPROFIT"],
            "归属于母公司净利润": ["sina_归属于母公司所有者的净利润", "ths_归属于母公司所有者的净利润", "em_PARENTNETPROFIT"],
            "每股收益": ["sina_基本每股收益(元)", "ths_基本每股收益", "em_BASICEPS"],
            "总资产": ["sina_资产总计", "ths_资产总计", "em_TOTALASSETS"],
            "总负债": ["sina_负债合计", "ths_负债合计", "em_TOTALLIABILITIES"],
            "净资产": ["sina_所有者权益(或股东权益)合计", "ths_所有者权益(或股东权益)合计", "em_TOTALEQUITY"],
            "资产负债率": ["sina_资产负债率(%)", "ths_资产负债率", "em_ASSETSLIABRATIO"],
            "货币资金": ["sina_货币资金", "ths_货币资金", "em_MONETARYFUNDS"],
            "应收账款": ["sina_应收账款", "ths_应收账款", "em_ACCOUNTS_RECE"],
            "应付账款": ["sina_应付账款", "ths_应付账款", "em_ACCOUNTS_PAYABLE"],
            "存货": ["sina_存货", "ths_存货", "em_INVENTORY"],
            "合同负债": ["sina_合同负债", "ths_合同负债", "em_CONTRACTLIAB"],
            "流动资产": ["sina_流动资产合计", "ths_流动资产合计", "em_TOTALCURRENTASSETS"],
            "流动负债": ["sina_流动负债合计", "ths_流动负债合计", "em_TOTALCURRENTLIAB"],
            "经营活动现金流量净额": ["sina_经营活动产生的现金流量净额", "ths_经营活动现金流量净额", "em_NETOPERATECASHFLOW"],
            "投资活动现金流量净额": ["sina_投资活动产生的现金流量净额", "ths_投资活动现金流量净额", "em_NETINVESTCASHFLOW"],
            "筹资活动现金流量净额": ["sina_筹资活动产生的现金流量净额", "ths_筹资活动现金流量净额", "em_NETFINACASHFLOW"],
            "现金净增加额": ["sina_现金及现金等价物净增加额", "ths_现金及现金等价物净增加额", "em_NETCASHEQUI"],
            "资本性支出": ["sina_购建固定资产、无形资产和其他长期资产支付的现金", "ths_购建固定资产、无形资产和其他长期资产支付的现金", "em_BUILDFIXASSETPAY"],
            "净资产收益率": ["ths_净资产收益率", "em_ROE"],
            "销售毛利率": ["sina_销售毛利率(%)", "ths_销售毛利率", "em_GROSSRATEMARGIN"],
            "销售净利率": ["sina_销售净利率(%)", "ths_销售净利率", "em_NETPROFITMARGIN"],
            "营业成本": ["sina_营业成本", "ths_营业成本", "em_OPERATECOST"],
            "现金股利": ["sina_分配股利、利润或偿付利息支付的现金", "ths_分配股利、利润或偿付利息支付的现金", "em_DIVIPROFITORINTERESTPAY"],
        }

        self.allowed_fields = set()
        for source_fields in self.core_indicators.values():
            self.allowed_fields.update(source_fields)

        self.business_description = self._get_business_description()
        self.historical_data = {}
        logger.info("AkshareFinancialDataFetcher初始化完成")

    def get_financial_data(self):
        """
        获取完整的财务数据
        
        返回:
        结构化财务数据字典，包含:
          - 公司基本信息
          - 财务报告(按报告期)
          - 主要股东信息
        """
        financial_data = {
            "股票代码": self.stock_code,
            "公司基本信息": {},
            "财务报告": {},
            "主要股东": {}
        }

        try:
            financial_data["公司基本信息"] = self.business_description

            sina_data = self._get_sina_data()
            self._merge_data(financial_data["财务报告"], sina_data)

            ths_data = self._get_ths_data()
            self._merge_data(financial_data["财务报告"], ths_data)

            em_data = self._get_em_data()
            self._merge_data(financial_data["财务报告"], em_data)

            ths_key_indicators = self._get_ths_key_indicators()
            self._merge_data(financial_data["财务报告"], ths_key_indicators)

            self._cross_validate_data(financial_data["财务报告"])

            self._calculate_derived_indicators(financial_data["财务报告"])

            self._filter_report_periods(financial_data["财务报告"])

            financial_data["主要股东"] = self._get_main_stock_holders()

            logger.info("财务数据获取完成")
            return financial_data
        except Exception as e:
            logger.exception(f"获取财务数据失败: {str(e)}")
            raise

    def _get_business_description(self):
        """获取公司主营业务描述"""
        logger.info("获取同花顺的主营业务描述...")
        try:
            business_df = ak.stock_zyjs_ths(symbol=self.ths_symbol)
            if not business_df.empty:
                result = {
                    "主营业务": business_df.iloc[0].get("主营业务", "未获取到主营业务信息"),
                    "主要产品": business_df.iloc[0].get("产品名称", "未获取到产品名称信息"),
                }
                logger.debug(f"主营业务: {result.get('主营业务', '未知')}")
                return result
            else:
                logger.warning("同花顺主营业务描述返回空数据")
        except Exception as e:
            logger.error(f"获取同花顺主营业务描述失败: {str(e)}")

        return {
            "主营业务": "未获取到主营业务信息",
            "主要产品": "未获取到产品名称信息",
        }

    def _get_sina_data(self):
        """从新浪财经获取财务数据"""
        data = {}
        try:
            for report_type in ["资产负债表", "利润表", "现金流量表"]:
                report_df = ak.stock_financial_report_sina(stock=self.stock_code, symbol=report_type)
                for _, row in report_df.iterrows():
                    report_date = row["报告日"]
                    if report_date not in data:
                        data[report_date] = {}

                    for col in report_df.columns:
                        if col == "报告日":
                            continue

                        field_name = f"sina_{col}"
                        if field_name in self.allowed_fields and pd.notna(row[col]):
                            value = row[col]
                            if isinstance(value, str) and value.endswith('元'):
                                try:
                                    value = float(value.replace('元', '').replace(',', ''))
                                except:
                                    pass
                            data[report_date][field_name] = value
            logger.info(f"从新浪获取 {len(data)} 个报告期数据")
        except Exception as e:
            logger.error(f"从新浪获取数据失败: {str(e)}")
        return data

    def _get_ths_data(self):
        """从同花顺获取财务数据"""
        data = {}
        try:
            for report_type, func_name in [("资产负债表", "stock_financial_debt_ths"),
                                           ("利润表", "stock_financial_benefit_ths"),
                                           ("现金流量表", "stock_financial_cash_ths")]:
                for period in ["按报告期", "按年度"]:
                    func = getattr(ak, func_name)
                    report_df = func(symbol=self.ths_symbol, indicator=period)
                    if "报告期" in report_df.columns:
                        for _, row in report_df.iterrows():
                            report_date = self._normalize_report_date(row["报告期"])
                            if report_date not in data:
                                data[report_date] = {}

                            for col in report_df.columns:
                                if col == "报告期":
                                    continue

                                field_name = f"ths_{col}"
                                if field_name in self.allowed_fields and pd.notna(row[col]):
                                    value = self._convert_value(row[col])
                                    data[report_date][field_name] = value
            logger.info(f"从同花顺获取 {len(data)} 个报告期数据")
        except Exception as e:
            logger.error(f"从同花顺获取数据失败: {str(e)}")
        return data

    def _get_em_data(self):
        """从东方财富获取财务数据"""
        data = {}
        try:
            for func in [ak.stock_balance_sheet_by_report_em, ak.stock_balance_sheet_by_yearly_em]:
                report_df = func(symbol=self.em_symbol)
                date_col = "REPORT_DATE" if "REPORT_DATE" in report_df.columns else "REPORTDATE"
                for _, row in report_df.iterrows():
                    report_date = self._normalize_report_date(row[date_col])
                    if report_date not in data:
                        data[report_date] = {}

                    for col in report_df.columns:
                        if col == date_col:
                            continue

                        field_name = f"em_{col}"
                        if field_name in self.allowed_fields and pd.notna(row[col]):
                            data[report_date][field_name] = row[col]

            for func in [ak.stock_profit_sheet_by_report_em, ak.stock_profit_sheet_by_yearly_em]:
                report_df = func(symbol=self.em_symbol)
                date_col = "REPORT_DATE"
                for _, row in report_df.iterrows():
                    report_date = self._normalize_report_date(row[date_col])
                    if report_date not in data:
                        data[report_date] = {}

                    for col in report_df.columns:
                        if col == date_col:
                            continue

                        field_name = f"em_{col}"
                        if field_name in self.allowed_fields and pd.notna(row[col]):
                            data[report_date][field_name] = row[col]

            for func in [ak.stock_cash_flow_sheet_by_report_em, ak.stock_cash_flow_sheet_by_yearly_em]:
                report_df = func(symbol=self.em_symbol)
                date_col = "REPORT_DATE"
                for _, row in report_df.iterrows():
                    report_date = self._normalize_report_date(row[date_col])
                    if report_date not in data:
                        data[report_date] = {}

                    for col in report_df.columns:
                        if col == date_col:
                            continue

                        field_name = f"em_{col}"
                        if field_name in self.allowed_fields and pd.notna(row[col]):
                            data[report_date][field_name] = row[col]

            logger.info(f"从东方财富获取 {len(data)} 个报告期数据")
        except Exception as e:
            logger.error(f"从东方财富获取数据失败: {str(e)}")
        return data

    def _get_ths_key_indicators(self):
        """从同花顺获取关键财务指标"""
        data = {}
        try:
            for period in ["按报告期", "按年度"]:
                report_df = ak.stock_financial_abstract_ths(symbol=self.ths_symbol, indicator=period)
                if "报告期" in report_df.columns:
                    for _, row in report_df.iterrows():
                        report_date = self._normalize_report_date(row["报告期"])
                        if report_date not in data:
                            data[report_date] = {}

                        col_mapping = {
                            "流动比率": "流动比率",
                            "速动比率": "速动比率",
                            "应收账款周转天数": "应收账款周转天数",
                            "存货周转天数": "存货周转天数",
                            "总资产周转率": "总资产周转率",
                            "净资产收益率": "净资产收益率",
                            "销售毛利率": "销售毛利率",
                            "销售净利率": "销售净利率"
                        }

                        for col in report_df.columns:
                            if col == "报告期":
                                continue

                            if col in col_mapping:
                                field_name = f"ths_{col_mapping[col]}"
                                if field_name in self.allowed_fields and pd.notna(row[col]):
                                    value = self._convert_value(row[col])
                                    data[report_date][field_name] = value
            logger.info(f"获取 {len(data)} 个关键指标报告期")
        except Exception as e:
            logger.error(f"从同花顺获取关键指标失败: {str(e)}")
        return data

    def _get_main_stock_holders(self):
        """获取主要股东信息"""
        holders_data = {}
        try:
            if self.stock_number.startswith(('6', '9')):
                symbol = f"sh{self.stock_number}"
            else:
                symbol = f"sz{self.stock_number}"

            current_date = datetime.now()
            year = current_date.year
            month = current_date.month

            if 1 <= month <= 3:
                report_dates = [f"{year-1}0930", f"{year-1}0630", f"{year-1}0331", f"{year-2}1231"]
            elif 4 <= month <= 6:
                report_dates = [f"{year}0331", f"{year-1}1231", f"{year-1}0930", f"{year-1}0630"]
            elif 7 <= month <= 9:
                report_dates = [f"{year}0630", f"{year}0331", f"{year-1}1231", f"{year-1}0930"]
            else:
                report_dates = [f"{year}0930", f"{year}0630", f"{year}0331", f"{year-1}1231"]

            for date_str in report_dates:
                try:
                    holders_df = ak.stock_gdfx_top_10_em(symbol=symbol, date=date_str)
                    report_date = self._normalize_report_date(date_str)
                    holders_data[report_date] = []

                    for _, row in holders_df.iterrows():
                        holder_info = {
                            "股东名称": row["股东名称"],
                            "持股比例(%)": row["占总股本持股比例"]
                        }
                        holders_data[report_date].append(holder_info)
                except Exception as e:
                    logger.warning(f"报告期 {date_str} 数据获取失败: {str(e)}")

            logger.info(f"获取 {len(holders_data)} 个股东报告期数据")
        except Exception as e:
            logger.error(f"获取主要股东数据失败: {str(e)}")

        return holders_data

    def _normalize_report_date(self, date_str):
        """标准化报告日期格式"""
        if not isinstance(date_str, str):
            date_str = str(date_str)

        try:
            date_formats = [
                "%Y%m%d",
                "%Y-%m-%d",
                "%Y/%m/%d",
                "%Y-%m-%d %H:%M:%S",
                "%Y"
            ]

            for fmt in date_formats:
                try:
                    dt = datetime.strptime(date_str, fmt)
                    if fmt == "%Y":
                        result = dt.strftime("%Y1231")
                        logger.debug(f"标准化 {date_str} -> {result} (年度报告)")
                        return result
                    result = dt.strftime("%Y%m%d")
                    logger.debug(f"标准化 {date_str} -> {result} (格式 {fmt})")
                    return result
                except ValueError:
                    continue

            if "Q" in date_str:
                year, quarter = date_str.split("Q")
                quarter_end = {"1": "0331", "2": "0630", "3": "0930", "4": "1231"}.get(quarter, "1231")
                result = f"{year}{quarter_end}"
                logger.debug(f"标准化季度报告 {date_str} -> {result}")
                return result

            if "年" in date_str and ("季" in date_str or "度" in date_str):
                year = date_str.split("年")[0]
                if "第一" in date_str or "1" in date_str:
                    result = f"{year}0331"
                elif "第二" in date_str or "2" in date_str or "上半" in date_str:
                    result = f"{year}0630"
                elif "第三" in date_str or "3" in date_str:
                    result = f"{year}0930"
                elif "第四" in date_str or "4" in date_str or "全年" in date_str:
                    result = f"{year}1231"
                else:
                    result = date_str
                logger.debug(f"标准化中文报告 {date_str} -> {result}")
                return result

            logger.debug(f"无法识别的日期格式: {date_str}")
            return date_str
        except Exception as e:
            logger.error(f"标准化报告日期失败: {date_str} - {str(e)}")
            return date_str

    def _convert_value(self, value):
        """转换财务值格式"""
        if isinstance(value, str):
            if "%" in value:
                try:
                    return float(value.replace("%", "")) / 100
                except:
                    pass

            if "亿" in value:
                try:
                    return float(value.replace("亿", "")) * 100000000
                except:
                    pass
            elif "万" in value:
                try:
                    return float(value.replace("万", "")) * 10000
                except:
                    pass

            if "," in value:
                try:
                    return float(value.replace(",", ""))
                except:
                    pass

            try:
                return float(value)
            except:
                pass

        return value

    def _merge_data(self, main_dict, new_dict):
        """合并不同数据源的数据"""
        logger.debug(f"合并数据: 现有 {len(main_dict)} 个报告期，新增 {len(new_dict)} 个报告期")
        for date, data in new_dict.items():
            if date in main_dict:
                main_dict[date].update(data)
            else:
                main_dict[date] = data
        logger.debug(f"合并后报告期总数: {len(main_dict)}")

    def _cross_validate_data(self, data_dict):
        """交叉验证不同数据源的数据"""
        logger.info(f"交叉验证 {len(data_dict)} 个报告期数据")
        for report_date, report_data in data_dict.items():
            validated_data = {}

            for indicator_name, source_fields in self.core_indicators.items():
                values = []
                sources = []

                for field in source_fields:
                    if field in report_data:
                        value = report_data[field]
                        if not isinstance(value, (int, float)):
                            try:
                                value = float(value)
                            except:
                                continue

                        values.append(value)
                        sources.append(field.split("_")[0])

                if not values:
                    continue

                min_val = min(values)
                max_val = max(values)
                avg_val = sum(values) / len(values)

                if avg_val != 0:
                    max_diff = abs(max_val - min_val) / abs(avg_val)
                else:
                    max_diff = 0

                if max_diff < 0.05:
                    validated_data[indicator_name] = avg_val
                    validated_data[f"{indicator_name}_数据来源"] = ",".join(sources)

            data_dict[report_date] = validated_data

        logger.info(f"交叉验证后保留 {len(data_dict)} 个报告期")

    def _calculate_derived_indicators(self, data_dict):
        """计算衍生财务指标"""
        logger.info(f"为 {len(data_dict)} 个报告期计算衍生指标")
        self.historical_data = data_dict.copy()

        for report_date, report_data in data_dict.items():
            try:
                current_assets = report_data["流动资产"]
                current_liabilities = report_data["流动负债"]
                inventory = report_data["存货"]
                current_ratio = current_assets / current_liabilities
                quick_ratio = (current_assets - inventory) / current_liabilities
                report_data["流动比率"] = current_ratio
                report_data["速动比率"] = quick_ratio
            except KeyError as e:
                pass

            try:
                operating_cash_flow = report_data["经营活动现金流量净额"]
                capital_expenditure = report_data["资本性支出"]
                free_cash_flow = operating_cash_flow - capital_expenditure
                report_data["自由现金流"] = free_cash_flow
            except KeyError as e:
                pass

            try:
                prev_date = self._find_previous_report_date(report_date)
                if prev_date and prev_date in self.historical_data:
                    prev_data = self.historical_data[prev_date]
                    revenue = report_data["营业总收入"]
                    ar_current = report_data["应收账款"]
                    ar_prev = prev_data["应收账款"]
                    avg_ar = (ar_current + ar_prev) / 2
                    if avg_ar != 0:
                        ar_turnover_days = 365 / (revenue / avg_ar)
                        report_data["应收账款周转天数"] = ar_turnover_days

                    cogs = report_data["营业成本"]
                    inventory_current = report_data["存货"]
                    inventory_prev = prev_data["存货"]
                    avg_inventory = (inventory_current + inventory_prev) / 2
                    if avg_inventory != 0:
                        inventory_turnover_days = 365 / (cogs / avg_inventory)
                        report_data["存货周转天数"] = inventory_turnover_days

                    total_assets_current = report_data["总资产"]
                    total_assets_prev = prev_data["总资产"]
                    avg_total_assets = (total_assets_current + total_assets_prev) / 2
                    if avg_total_assets != 0:
                        asset_turnover = revenue / avg_total_assets
                        report_data["总资产周转率"] = asset_turnover
            except Exception as e:
                logger.warning(f"{report_date}: 计算周转率指标失败: {str(e)}")

            try:
                net_profit_attributable = report_data["归属于母公司净利润"]
                cash_dividends = report_data["现金股利"]
                if net_profit_attributable != 0:
                    dividend_payout_ratio = cash_dividends / net_profit_attributable
                    report_data["股息支付率"] = dividend_payout_ratio

                if self.stock_price and net_profit_attributable != 0:
                    total_shares = report_data.get("总股本", None)
                    if total_shares is None:
                        total_shares = report_data.get("ths_总股本", report_data.get("em_TOTAL_SHARE", None))

                    if total_shares and total_shares != 0:
                        dividend_per_share = cash_dividends / total_shares
                        dividend_yield = dividend_per_share / self.stock_price
                        report_data["股息收益率"] = dividend_yield
            except KeyError as e:
                pass

    def _find_previous_report_date(self, current_date):
        """查找前一个报告期日期"""
        def date_key(d):
            year = int(d[:4])
            month_day = d[4:]
            if month_day == "1231":
                return (year, 4)
            quarter_map = {"0331": 1, "0630": 2, "0930": 3}
            return (year, quarter_map.get(month_day, 0))

        dates = list(self.historical_data.keys())
        sorted_dates = sorted(dates, key=date_key)

        try:
            current_idx = sorted_dates.index(current_date)
            if current_idx > 0:
                return sorted_dates[current_idx - 1]
        except ValueError:
            pass

        return None

    def _filter_report_periods(self, data_dict):
        """过滤报告期，保留最近的8个季度报告"""
        current_year = datetime.now().year
        current_month = datetime.now().month

        required_periods = [
            f"{current_year}0331", f"{current_year}0630", f"{current_year}0930",
            f"{current_year-1}0630", f"{current_year-1}0930",
            f"{current_year-2}0630", f"{current_year-2}1231",
            f"{current_year-1}1231"
        ]

        keys_to_remove = []
        for report_date in data_dict.keys():
            std_date = self._normalize_report_date(report_date)
            if std_date not in required_periods:
                keys_to_remove.append(report_date)

        for key in keys_to_remove:
            del data_dict[key]

        logger.info(f"过滤后保留 {len(data_dict)} 个报告期")
